package persist

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.junit.{After, Before, Test}

/**
 * 除了窄依赖的依赖都是宽依赖
 */
class Driver {

	val conf: SparkConf = new SparkConf().setAppName("Serializ").setMaster("local[3]")
	var sc: SparkContext = null
	var outpath: String = "out"
	import util.MyPredef._

	@Before
	def init() {
		sc = new SparkContext(conf)
		outpath.delete()
	}

	@After
	def after() {
		sc.stop()
	}

	@Test
	def main1(): Unit = {
		val word = sc.textFile("in").flatMap(_.split(" "))

		val res1 =  word.map(_.toString + "\t" + System.currentTimeMillis)
		val res2 =  word.map(_.toString + "\t" + System.currentTimeMillis).cache

		res1.foreach(println) // 时间戳不一样
		println("------------")
		res2.foreach(println) // 行动算子执行才会触发checkpoint

		println(res1.toDebugString)
		println(res2.toDebugString) // CachedPartitions, 内存中会记住血缘，磁盘会忘记
	}

	@Test
	def checkpointTest(): Unit = {
		sc.setCheckpointDir("cp")
		val word = sc.textFile("in").flatMap(_.split(" "))

		val res =  word.map((_, 1)).reduceByKey(_+_)
		res.checkpoint()

		res.foreach(println) // 行动算子才会触发 checkpoint
		println(res.toDebugString) // CachedPartitions, 内存中会记住血缘，磁盘会忘记

	}
}
